package com.burges.net.state.managedKeyedState

import org.apache.flink.api.common.functions.RichFlatMapFunction
import org.apache.flink.api.common.state.{StateTtlConfig, ValueState, ValueStateDescriptor}
import org.apache.flink.api.common.time.Time
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.util.Collector

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/5 
  * 描述     通过创建ValueState来获取指标的最小值
  */
object ValueStateDemo {

	def main(args: Array[String]): Unit = {
		val environment = StreamExecutionEnvironment.getExecutionEnvironment
		// 创建元素数据集
		val inputStream: DataStream[(Int, Long)] = environment.fromElements((2, 2L), (4, 1L), (5, 4L))
		inputStream.keyBy(_._1).flatMap{
			// 定义和创建RichFlatMapFunction，第一个参数是输入类型，第二个参数为输出数据类型
			new RichFlatMapFunction[(Int, Long), (Int, Long, Long)] {
				private var leastValueState: ValueState[Long] = _
				override def flatMap(t: (Int, Long), collector: Collector[(Int, Long, Long)]): Unit = {
					// 通过Value方法从leastValueState中的最小值
					val leastValue: Long = leastValueState.value()
					if(t._2 > leastValue){
						collector.collect((t._1, t._2, leastValue))
					}else {
						// 如果当前指标小于最小值，则更新状态中的最小值
						leastValueState.update(t._2)
						// 将当前数据中的指标作为最小值
						collector.collect((t._1, t._2, t._2))
					}
				}

				override def open(parameters: Configuration): Unit = {
					// 创建ValueStateDescriptor，定义状态名称为leastValue，并指定数据类型
					val leastValueStateDescriptor: ValueStateDescriptor[Long] = new ValueStateDescriptor[Long]("leastValue", classOf[Long])
					// 通过getRuntimeContext.getState获取State  获取状态数据
					leastValueState = getRuntimeContext.getState(leastValueStateDescriptor)
				}
			}
		}

		//State生命周期
		//创建StateTtlConfig
		val stateTtlConfig: StateTtlConfig = StateTtlConfig
				//设定TTL时长为10s  设置时间参数是必须的之外，其他参数都是可选的或者使用默认值
				.newBuilder(Time.seconds(10))
				//指定TTL刷新时只对创建和写入操作有效
				/**
				  * StateTtlConfig.UpdateType.OnCreateAndWrite 仅在创建和写入时更新TTL
				  * StateTtlConfig.UpdateType.OnReadAndWrite   所有读与写操作都更新TTL
				  *
				  * 如果某个状态指标一直不被使用或者更新，则永远不会触发对该状态数据的清理操作，可以使用StateTtlConfig.cleanupFullSnapshot设定触发StateSnapshot的时候清理状态数据
				  *    需要注意这个配置不适合用于RocksDB做增量Checkpointing的操作
				  */
				.setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
				//指定状态可见性为永远不反馈过期数据
				/**
				  * 设定状态的可见性，根据过期数据是否被清理来确定是否返回状态数据
				  * StateTtlConfig.StateVisibility.NeverReturnExpired          状态数据过期就不会返回(默认)
				  * StateTtlConfig.StateVisibility.ReturnExpiredIfNotCleanedUp 状态数据即使过期但没有被清理依然返回
				  */
				.setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
				.build()
		//创建valueStateDescriptor
		val valueStateDescriptor: ValueStateDescriptor[Long] = new ValueStateDescriptor[Long]("valueState", classOf[Long])
		valueStateDescriptor.enableTimeToLive(stateTtlConfig)

	}


}
